package mspbots.cw.client.core.client;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.log4j.Log4j2;
import mspbots.cw.common.RabbitKeys;
import mspbots.cw.common.ResultSync;
import mspbots.cw.common.annotation.ApiException;
import org.springframework.amqp.rabbit.core.RabbitMessagingTemplate;
import org.springframework.hateoas.Link;
import org.springframework.hateoas.LinkRelation;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
import org.springframework.util.MultiValueMap;
import org.springframework.util.ObjectUtils;
import org.springframework.web.reactive.function.client.ClientResponse;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.util.UriComponentsBuilder;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.net.URI;
import java.net.URLDecoder;
import java.nio.charset.Charset;
import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * com.mspbots.sync.wise.client.AutoPagerClient
 *
 * @author Alex bob(https://github.com/vnobo)
 * @date Created by 2020/7/3
 */
@Log4j2
@Service
public class CwWebClient {

    private final static String CW_CLIENT_ID = "b9feb0a1-787c-46b9-9d85-b6f60e2b927a";

    private final WebClient webClient;
    private final ObjectMapper objectMapper;
    private final RabbitMessagingTemplate rabbitMessagingTemplate;

    public CwWebClient(WebClient.Builder builder,
                       RabbitMessagingTemplate rabbitMessagingTemplate,
                       ObjectMapper objectMapper) {
        this.rabbitMessagingTemplate = rabbitMessagingTemplate;
        this.objectMapper = objectMapper;

        this.webClient = builder
                .defaultHeader("Accept", "application/vnd.connectwis.com+json")
                .defaultHeader("ClientId", CW_CLIENT_ID)
                .build();
    }

    public Mono<JsonNode> put(String path, Object body) {
        return Mono.deferContextual(ctx -> {
            ClientProperties properties = ctx.get("properties");
            ResultSync resultSync = ctx.getOrDefault("resultSync", ResultSync.withDefault(path));
            assert resultSync != null;
            resultSync.setElapsedTime(System.currentTimeMillis());
            resultSync.setTenantId(properties.getTenantId());
            resultSync.setParams(this.objectMapper.convertValue(body, JsonNode.class));
            return this.webClient.put().uri(UriComponentsBuilder.newInstance().scheme("https")
                    .host(properties.getSite()).path("/v4_6_release/apis/3.0/")
                    .path(path).build().toUri())
                    .headers(headers -> headers.setBasicAuth(properties.getAppId() + "+" + properties.getPublicKey(),
                            properties.getPrivateKey()))
                    .bodyValue(body)
                    .exchangeToMono(clientResponse -> this.handlerRequestResult(clientResponse, resultSync))
                    .flatMap(responseEntity -> {
                        if (ObjectUtils.isEmpty(responseEntity.getBody())) {
                            return Mono.empty();
                        }
                        return Mono.just(responseEntity.getBody());
                    });
        });
    }

    public Mono<Boolean> delete(String path, Map<String, ?> pathParams) {
        return Mono.deferContextual(ctx -> {
            ClientProperties properties = ctx.get("properties");
            ResultSync resultSync = ctx.getOrDefault("resultSync", ResultSync.withDefault(path));
            assert resultSync != null;
            resultSync.setElapsedTime(System.currentTimeMillis());
            resultSync.setTenantId(properties.getTenantId());
            resultSync.setParams(this.objectMapper.convertValue(pathParams, JsonNode.class));
            return this.webClient.delete().uri(UriComponentsBuilder.newInstance().scheme("https")
                    .host(properties.getSite()).path("/v4_6_release/apis/3.0/").path(path).build(pathParams))
                    .headers(headers -> headers.setBasicAuth(properties.getAppId() + "+" + properties.getPublicKey(),
                            properties.getPrivateKey()))
                    .exchangeToMono(clientResponse -> this.handlerRequestResult(clientResponse, resultSync))
                    .map(clientResponse -> clientResponse.getStatusCode().is2xxSuccessful());
        });
    }

    public Mono<JsonNode> post(String path, Object body) {
        return Mono.deferContextual(ctx -> {
            ClientProperties properties = ctx.get("properties");
            ResultSync resultSync = ctx.getOrDefault("resultSync", ResultSync.withDefault(path));
            assert resultSync != null;
            resultSync.setElapsedTime(System.currentTimeMillis());
            resultSync.setTenantId(properties.getTenantId());
            resultSync.setParams(this.objectMapper.convertValue(body, JsonNode.class));
            return this.webClient.post().uri(UriComponentsBuilder.newInstance().scheme("https")
                    .host(properties.getSite()).path("/v4_6_release/apis/3.0/")
                    .path(path).build().toUri())
                    .headers(headers -> headers.setBasicAuth(properties.getAppId() + "+" + properties.getPublicKey(),
                            properties.getPrivateKey()))
                    .bodyValue(body)
                    .exchangeToMono(clientResponse -> this.handlerRequestResult(clientResponse, resultSync))
                    .flatMap(responseEntity -> {
                        if (ObjectUtils.isEmpty(responseEntity.getBody())) {
                            return Mono.empty();
                        }
                        return Mono.just(responseEntity.getBody());
                    });
        });
    }

    public Mono<JsonNode> get(String path, MultiValueMap<String, String> queryParams) {
        return this.getEntity(path, queryParams)
                .flatMap(responseEntity -> {
                    if (responseEntity.getStatusCode().isError() || ObjectUtils.isEmpty(responseEntity.getBody())) {
                        return Mono.empty();
                    }
                    return Mono.just(responseEntity.getBody());
                });
    }

    public Mono<ResponseEntity<JsonNode>> getEntity(String path, MultiValueMap<String, String> queryParams) {
        return Mono.deferContextual(ctx -> {
            ClientProperties properties = ctx.get("properties");
            URI uri = UriComponentsBuilder.newInstance().scheme("https")
                    .host(properties.getSite()).path("/v4_6_release/apis/3.0/")
                    .path(path).queryParams(queryParams).build().toUri();
            return this.requestEntity(properties, uri);
        });
    }

    public Mono<ResponseEntity<JsonNode>> requestEntity(ClientProperties properties, URI uri) {
        return this.webClient.get().uri(uri)
                .headers(httpHeaders -> httpHeaders.setBasicAuth(properties.getAppId() +
                        "+" + properties.getPublicKey(), properties.getPrivateKey()))
                .exchangeToMono(clientResponse -> {
                    if (clientResponse.statusCode().isError()) {
                        return clientResponse.createException()
                                .flatMap(errNode -> Mono.error(ApiException
                                        .withMsg(clientResponse.statusCode(), errNode.getResponseBodyAsString())));
                    }
                    return clientResponse.toEntity(JsonNode.class);
                });
    }

    public Flux<JsonNode> fetch(String path, MultiValueMap<String, String> queryParams, Map<String, String> pathParams) {
        return Mono.deferContextual(ctx -> Mono.just((ClientProperties) ctx.get("properties")))
                .flatMapMany(properties -> this.fetch(UriComponentsBuilder.newInstance().scheme("https")
                        .host(properties.getSite()).path("/v4_6_release/apis/3.0/")
                        .path(path).queryParams(queryParams).build(pathParams)));
    }

    public Flux<JsonNode> fetch(String path, MultiValueMap<String, String> queryParams) {
        return Mono.deferContextual(ctx -> Mono.just((ClientProperties) ctx.get("properties")))
                .flatMapMany(properties -> this.fetch(UriComponentsBuilder.newInstance().scheme("https")
                        .host(properties.getSite()).path("/v4_6_release/apis/3.0/")
                        .path(path).queryParams(queryParams).build().toUri()))
                .flatMap(Flux::fromIterable);
    }

    public Flux<JsonNode> fetch(URI uri) {
        return this.request(uri)
                .expand(this::autoNextPage)
                .delayElements(Duration.ofMillis(200))
                .flatMap(responseEntity -> {
                    if (ObjectUtils.isEmpty(responseEntity.getBody())) {
                        return Flux.empty();
                    }
                    return Flux.just(responseEntity.getBody());
                })
                .limitRate(50, 25);
    }

    private Mono<ResponseEntity<JsonNode>> autoNextPage(ResponseEntity<JsonNode> resp) {
        List<String> links = resp.getHeaders().getValuesAsList(HttpHeaders.LINK);
        URI newUrl = links.parallelStream().map(Link::valueOf)
                .filter(link -> link.hasRel(LinkRelation.of("next")))
                .map(link -> UriComponentsBuilder.fromHttpUrl(URLDecoder.decode(link.getHref(), Charset.defaultCharset()))
                        .build().toUri())
                .findAny().orElse(null);

        if (ObjectUtils.isEmpty(newUrl)) {
            return Mono.empty();
        }
        log.debug("===Request next url [{}]===", newUrl.toString());
        return this.request(newUrl);
    }

    private Mono<ResponseEntity<JsonNode>> request(URI uri) {
        log.debug("Request url {}", uri);
        return Mono.deferContextual(ctx -> {
            ClientProperties properties = ctx.get("properties");
            ResultSync resultSync = ctx.getOrDefault("resultSync", ResultSync.withDefault(uri.getPath()));
            assert resultSync != null;
            resultSync.setTenantId(properties.getTenantId());
            resultSync.setParams(this.objectMapper
                    .convertValue(UriComponentsBuilder.fromUri(uri).build().getQueryParams(), JsonNode.class));
            resultSync.setElapsedTime(System.currentTimeMillis());
            return this.webClient.get().uri(uri)
                    .headers(headers -> headers.setBasicAuth(properties.getAppId() +
                                    "+" + properties.getPublicKey(),
                            properties.getPrivateKey()))
                    .exchangeToMono(clientResponse -> this.handlerRequestResult(clientResponse, resultSync))
                    .timeout(Duration.ofMinutes(2));
        });
    }


    private Mono<ResponseEntity<JsonNode>> handlerRequestResult(ClientResponse response, ResultSync resultSync) {

        resultSync.setStatus(response.statusCode());
        resultSync.setElapsedTime(System.currentTimeMillis() - resultSync.getElapsedTime());

        if (response.statusCode().isError()) {
            var bodyMono = response.headers().contentType().stream()
                    .allMatch(MediaType.APPLICATION_JSON::isCompatibleWith) ? response.bodyToMono(JsonNode.class)
                    .defaultIfEmpty(this.objectMapper.createObjectNode()
                            .put("error", "This request API error.")
                            .putPOJO("headers", response.headers().asHttpHeaders())) :
                    response.bodyToMono(String.class);
            return bodyMono.doOnNext(errNode -> {
                resultSync.setMsg(errNode);
                log.debug("sync [{}] data error action {}, elapsedTime {}, status: {}",
                        resultSync.getTenantId(), resultSync.getAction(),
                        resultSync.getElapsedTime(), resultSync.getStatus());

                this.rabbitMessagingTemplate.convertAndSend(RabbitKeys.BASE_EXCHANGE_NAME,
                        RabbitKeys.LOGS_TASK_QUEUE_KEY, resultSync);
            }).flatMap(errNode -> Mono.error(ApiException.withMsg(response.statusCode(), errNode)));

        }

        long contentLength = response.headers().contentLength().orElse(0);
        int count = contentLength < 1024 ? 1 : (int) contentLength / 1024;
        resultSync.setCount(count);

        log.debug("Sync [{}] data success, action [{}], elapsedTime [{}], count [{}]",
                resultSync.getTenantId(), resultSync.getAction(),
                resultSync.getElapsedTime(), resultSync.getCount());

        this.rabbitMessagingTemplate.convertAndSend(RabbitKeys.BASE_EXCHANGE_NAME,
                RabbitKeys.LOGS_TASK_QUEUE_KEY, resultSync);

        return response.toEntity(JsonNode.class);
    }


}
